package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/2/6
 *
 *  TVFwindow 对比 Group Window:
 *          ①提供性能优化措施
 *          ②支持 GroupingSets 语法
 *                  select xx from  xx group by a,b
 *                  grouping sets ( (a,b), (a) , (b) , () )  hive,flink支持
 *                  cube      ck,hive,flink支持
 *                  rollup     ck,hive,flink支持
 *                  withTotal   只有ck支持
 *          ③支持窗口中求TopN
 *
 *    缺陷: 目前1.13仅支持
 *      Tumble Windows
 *      Hop Windows
 *      Cumulate Windows   累积窗口
 *      Session Windows (will be supported soon)  目前不支持。
 */
public class Demo13_TVFWindowCumulate
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());


        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);


        /*
                只选取流中数据的某些属性，组成Table
         */
        Table table = tableEnvironment.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),$("et").rowtime()
        );

        tableEnvironment.createTemporaryView("t1",table);

        /*
               累积窗口的适用场景：
                例如每隔1h统计当天的累积PV
                    1点时，统计 [0,1]
                    2点时，统计 [0,1]+[1,2]
                    3点时，统计 [0,1]+[1,2]+[2,3]
                    ....
                    直到当天结束。

               定义两个变量:
                    step: 累积的步长  1h
                    maxSize: 累积的极限 24h。 作用当窗口运行到maxSize时，下一次运算就重置归零。

               --------------
                举例:  step:  2s
                      maxsize : 6s
         */
        String cuculate = "SELECT window_start, window_end,  SUM(vc) sumVC" +
            "  FROM TABLE(" +
            "    CUMULATE(TABLE t1, DESCRIPTOR(et), INTERVAL '2' SECONDS, INTERVAL '6' SECONDS))" +
            "  GROUP BY window_start, window_end  " ;


        //不支持session
        tableEnvironment.sqlQuery(cuculate)
                        .execute()
                        .print();




    }
}
